package com.wx.spider.java.listener;

import com.alibaba.fastjson.JSONObject;
import com.wx.spider.common.model.mq.GetListMessage;
import com.wx.spider.common.pojo.db.MqErrorLog;
import com.wx.spider.common.util.IdGenerateUtil;
import com.wx.spider.java.service.PMToJavaService;
import com.wx.spider.rocketmq.annotation.WsRocketMQConsumer;
import com.wx.spider.rocketmq.consumer.BaseRocketMQConsumer;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 * @author:feng
 * @create:2021-04-21 12:40
 */
@Component
@Slf4j
@WsRocketMQConsumer(consumerGroup = "ws-consumer-pm-to-java", topic = "ws-get-list", consumeThreadMin = 3)
public class PMToJavaConsumer extends BaseRocketMQConsumer {

    @Autowired
    private PMToJavaService pmToJavaService;

    @Override
    protected ConsumeConcurrentlyStatus consume(String tag, String topic, MessageExt messageExt) {

        int count = 0;
        try {

            String msg = new String(messageExt.getBody());

            log.info(messageExt.getTopic() + ":" + messageExt.getTags() + "-" + messageExt.getKeys() + "--> 消息:" + msg);
            GetListMessage getListMessage = null;

            try {
                getListMessage = JSONObject.parseObject(msg, GetListMessage.class);
            } catch (Exception e) {
                e.printStackTrace();
            }

            //获取重试次数
            count = messageExt.getReconsumeTimes();
            //这里设置重试大于3次 那么通过保存数据库 人工来兜底
            if (count > 3) {
                log.info("该消息已经重试3次,保存数据库。topic={},keys={},msg={}", messageExt.getTopic(), messageExt.getKeys(), messageExt.getBody());
                MqErrorLog mqErrorLog = new MqErrorLog();
                mqErrorLog.setId(IdGenerateUtil.get().nextId());
                mqErrorLog.setAddtime(new Date());
                mqErrorLog.setRecount(count);
                if(getListMessage != null){
                    mqErrorLog.setTagId(getListMessage.getNewsId());
                }
                mqErrorLog.setTopic(topic);
                pmToJavaService.saveMqErrorLog(mqErrorLog);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            Boolean handlerSuccess = pmToJavaService.getContent(getListMessage);

            //直接抛出异常
            if (!handlerSuccess) {
                throw new Exception("=======任务执行失败========");
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } catch (Exception e) {
            e.printStackTrace();
            log.info("消费失败，当前消费重试次数为 = {}", count);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }

}
